package de.preissler.selection;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SubSliceQuery;

public class WinnersCalc implements Winners
{
	private static final String SCF = "Quotes";
	private static final String CHANGE = "change";

	public Map<String, List<ChangeOfISIN>> getWinners(List<String> dates,
			Keyspace keyspace, Map<String, List<ChangeOfISIN>> result, int number)
	{
		//Rufe alle verfuegbaren ISIN's aus der Datenbank ab
		List<String> isins = getKeys(keyspace, SCF);
		
		//Fuer jedes Datum ein neue Liste von sortierten Wertpapieren
		for(String date : dates)
		{
			List<ChangeOfISIN> changes = new ArrayList<ChangeOfISIN>(114);
			
			//Speichere fuer jede ISIN die relative Aenderung
 			for(String isin : isins)
			{
				try
				{
					double change = getChange(isin, date, keyspace);
					changes.add(new ChangeOfISIN(change, Arrays.asList(isin)));
				}
				catch(Exception e)
				{}
			}
 			//Sortiere die Liste nach relativer Aenderung
 			Collections.sort(changes, Collections.reverseOrder());
 			
 			//Fuege die Liste der Map unter dem entsprechendem Datum ein
			result.put(date, changes.subList(0, number));
		}
		
		return result;
	}
	
	public double getChange(String isin, String date, Keyspace keyspace)
	{
		SubSliceQuery<String, String, String, Double> query = HFactory.createSubSliceQuery(keyspace, SS, SS, SS, DS);
		query.setColumnFamily(SCF);
		query.setKey(isin);
		query.setSuperColumn(date);
		query.setColumnNames(CHANGE);
		
		return query.execute().get().getColumnByName(CHANGE).getValue();
	}
	
	/*
	 * Holt alle ISINs raus
	 */
	public List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}

}
